package com.SparkCore.RDD.Operator.Transform

import org.apache.spark.{SparkConf, SparkContext}


/**
 * 将待处理的数据以分区为单位发送到计算节点进行处理，这里的处理是指可以进行任意的处理，哪怕是过滤数据。
 */
object Spark02_RDD_Operator_Transform {

  def main(args: Array[String]): Unit = {

    //创建环境
    val sparkConf = new SparkConf().setMaster("local[*]").setAppName("RDD")
    val sparkContext = new SparkContext(sparkConf)

    //创建RDD算子 -- map
    val rdd = sparkContext.makeRDD(List(1,2,3,4) , 2)

    /**
     * mapPartition: 可以以分区为单位进行数据转换操作
     *                但是会把整个分区的数据都加载到内存中进行引用
     *                如果处理完的数据是不会被释放掉的，存在对象的引用
     *                在内存较小，数据量较大的场合下容易出现内存溢出
     *
     * map 没有引用，来一条处理一条，处理完就进行释放，不会存在内存溢出的状态
     */
    val mpRDD = rdd.mapPartitions(
      iter => {
        iter.map(_ * 2)
      }
    )
    mpRDD.collect().foreach(println)

    //关闭环境
    sparkContext.stop()
  }
}
